package cn.tedu.test.charging.emqx.mqtt;

import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;

public class EmqxDemoConsumer implements MqttCallbackExtended {

    private MqttClient mqttClient;
    private MqttContext mqttContext;
    /*private OrderMapper orderMapper;
    private StockMapper stockMapper;
    private CartMapper cartMapper;
    private RedisTemplate redisTemplate;*/

    public EmqxDemoConsumer(MqttClient mqttClient,MqttContext mqttContext) {
        this.mqttClient = mqttClient;
        this.mqttContext = mqttContext;
       /* this.orderMapper = orderMapper;*/
    }

    /**
     * 连接成功 客户端mqttClient成功的建立了和emqx连接
     * @param reconnect mqttClient 回调消费传递的参数 告知 方法 本次连接建立是否支持重连
     * @param serverURI 服务端emqx地址
     */
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        //客户端对象mqttClient成功连接的时候 调用的 通过客户端mqttClient实现监听订阅
        try {
            mqttClient.subscribe("/demo/start/topic");
            mqttClient.subscribe("/demo/stop/topic");
        } catch (MqttException e) {
            System.out.println("订阅监听失败");
        }
    }

    /**
     * 连接断开 mqttClient断开和服务连接 回调的方法
     * 消费者使用的一些资源 消耗程序内容 在连接断开时候 释放避免因为这个消费者无法使用消费逻辑导致程序阻塞
     * @param cause the reason behind the loss of connection.
     */
    @Override
    public void connectionLost(Throwable cause) {
        System.out.println("连接断开");
    }

    /**
     * 消费者通过mqttClient监听订阅的主题 产生了消息,消息到达了当前消费对象
     * @param topic 消息主题来源 主题名称
     * @param message 消息对象
     * @throws Exception
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        //当emqx主题产生消息的时候 mqttClient 回调这个方法 告诉消费者消息对象和主题来源
        System.out.println("本次消息主题 "+topic);
        //主题不一样 消费逻辑不一样
        if (topic!=null&&topic.equals("/demo/start/topic")){
            //处理start逻辑
            byte[] payload = message.getPayload();//BODY
            System.out.println("本次消息BODY "+new String(payload));
            //调用容器bean对象orderMapper
            mqttContext.getOrderMapper().addOrder();
        }else if(topic!=null&&topic.equals("/demo/stop/topic")){
            //处理stop逻辑
            byte[] payload = message.getPayload();//BODY
            System.out.println("本次消息BODY "+new String(payload));
        }
    }

    /**
     * 消息投递 消息消费彻底完成
     * @param token message身份标识
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        System.out.println("消息消费完成");
    }
}
